package com.yuyunbo.hbase.autoconfigure;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.*;

/**
 * @Author: yuyunbo-1154363414@qq.com
 * @Date: 2020/5/8 20:47
 * @Description:
 */
@Configuration
@EnableConfigurationProperties(HbaseConfigProperties.class)
@ConditionalOnProperty(prefix = "hbase.client", value = "enable", matchIfMissing = true)
public class HbaseConfigBeanAutoConfigure {
    @Autowired
    private HbaseConfigProperties hbaseConfigProperties;

    @Bean(name = "hbaseThreadPool")
    public ThreadPoolExecutor initHbaseThreadPool() {
        BlockingQueue<Runnable> queue = this.createQueue(216);
        ThreadPoolExecutor taskExecutor = new ThreadPoolExecutor(8, 16, 60, TimeUnit.SECONDS, queue, (Runnable runnable) -> (
                                                                                                                                    new Thread(runnable, "hbase_thread_pool_" + runnable.hashCode())
        ));
        return taskExecutor;
    }

    private BlockingQueue<Runnable> createQueue(int queueCapacity) {
        return (BlockingQueue) (queueCapacity > 0 ? new LinkedBlockingQueue(queueCapacity) : new SynchronousQueue());
    }

    @Bean(name = "hbaseApiService")
    @ConditionalOnMissingBean(HbaseApiServiceImpl.class)
    public HbaseApiServiceImpl initHbaseApiService() {
        return new HbaseApiServiceImpl();
    }

    @Bean(name = "hbaseClient")
    @ConditionalOnMissingBean(HbaseClient.class)
    public HbaseClient initHbaseClient() {
        return new HbaseClient(hbaseConfigProperties.getZookeeperList(), hbaseConfigProperties.getZookeeperClientPort(), hbaseConfigProperties.getZookeeperZnodeParent(),
                                      hbaseConfigProperties.getUserName(), hbaseConfigProperties.getDbNameSpace(), hbaseConfigProperties.getZkSessionTimeout(), hbaseConfigProperties.getHbaseClientRetriesNumber());
    }
}
